package org.rcy.framework.mq.producer;

import org.rcy.framework.api.mq.AbstractMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * @author 吴康桥
 * @Description
 * @date 2022/12/5 9:40
 */


public class RabbitMQProducer<M extends AbstractMessage> implements  MessageProducer<M>{
	private final Logger logger = LoggerFactory.getLogger(this.getClass());
	private RabbitTemplate template;
	private RabbitAdmin admin;

	public RabbitMQProducer(RabbitTemplate template, RabbitAdmin admin) {
		this.template = template;
		this.admin = admin;
	}

	public void produce(M message) {
		String queueName = message.getQueueName();
		this.admin.declareQueue(new Queue(queueName));

		try {
			this.template.convertAndSend(queueName, message);
		} catch (Exception var4) {
			this.logger.error("Queue:{} error {}", new Object[]{queueName, var4.getMessage(), var4});
		}

	}

	@Override
	public void produce(M message, Long delay, TimeUnit timeUnit) {
		if (delay < 0L) {
			this.produce(message);
		}
		switch (timeUnit){
			case SECONDS: delay = delay * 1000; break;
			case MINUTES: delay = delay * 60000; break;
			default: break;
		}
		final String finalDelay = String.valueOf(delay);
		String queueName = message.getQueueName();
		Queue queue = new Queue(queueName);
		this.admin.declareQueue(queue);
		String exchangeName = "delay.common.x";
		DirectExchange exchange = new DirectExchange(exchangeName, true, false);
		this.admin.declareExchange(exchange);
		String delayQueueName = "delay." + queueName + "." + finalDelay;
		Map<String, Object> arguments = new HashMap();
		arguments.put("x-dead-letter-exchange", exchangeName);
		arguments.put("x-dead-letter-routing-key", message.getQueueName());
		Queue delayQueue = new Queue(delayQueueName, true, false, false, arguments);
		this.admin.declareQueue(delayQueue);
		this.admin.declareBinding(BindingBuilder.bind(queue).to(exchange).with(message.getQueueName()));
		MessagePostProcessor processor = new MessagePostProcessor() {
			public org.springframework.amqp.core.Message postProcessMessage(org.springframework.amqp.core.Message message) throws AmqpException {
				message.getMessageProperties().setExpiration(finalDelay);
				return message;
			}
		};
		this.template.convertAndSend(delayQueueName, message, processor);
	}
}
